package tasktracker.core.framework;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import tasktracker.abstraction.ITaskDataStore;
import tasktracker.abstraction.ITaskTracker;
import tasktracker.abstraction.ITaskTrackerContext;
import tasktracker.abstraction.ITaskTrackerFactory;
import tasktracker.abstraction.TaskInfo;
import tasktracker.abstraction.TaskProgressInfo;
import tasktracker.abstraction.TaskResultInfo;
import tasktracker.abstraction.TaskStatus;
import tasktracker.core.abstraction.FrameworkTrackedTask;
import tasktracker.core.abstraction.FrameworkTrackedTaskInfo;
import tasktracker.core.abstraction.TaskDataStoreProvider;
import tasktracker.core.abstraction.TaskStatusMessage;
import tasktracker.core.abstraction.TaskStoreProvider;
import tasktracker.core.abstraction.TaskTrackerProperties;
import tasktracker.core.helper.DateTimeHelper;

public class TaskTrackerContext {
	private Logger logger = LoggerFactory.getLogger(TaskTrackerContext.class);

	private int maxQueueSize;
	private int maxConcurrentSize;
	private boolean autoRestoreWaitingQueue;
	private TaskDataStoreProvider dataStoreProvider;
	private TaskStoreProvider taskStoreProvider;

	private Boolean shuttingdown = false;
	private boolean workingState = false;
	private ITaskTrackerFactory[] trackers;
	private Map<String, ITaskTracker> workingTrackers;
	private Map<String, TaskTrackerProperties> trackersProperties;

	private Semaphore dispatcherSemaphore;
	private BlockingQueue<FrameworkTrackedTask> waitingQueue;

	/**
	 * 初始化
	 */
	public synchronized void init() {
		if (workingState)
			throw new RuntimeException("TaskTrackerContext has already been initialized.");

		workingTrackers = new HashMap<String, ITaskTracker>();
		trackersProperties = new HashMap<String, TaskTrackerProperties>();
		for (ITaskTrackerFactory taskTrackerFactory : this.trackers) {
			if (taskTrackerFactory == null) {
				logger.warn("Null instance of ITaskTrackerFactory is provided");
				continue;
			}
			Map<String, ITaskTracker> trackers = taskTrackerFactory.getInstances();
			if (trackers == null) {
				logger.warn(taskTrackerFactory.getClass().getName() + ".getInstances returned null");
				continue;
			}
			for (Map.Entry<String, ITaskTracker> trackerEntry : trackers.entrySet()) {
				String name = trackerEntry.getKey();
				ITaskTracker tracker = trackerEntry.getValue();
				if (tracker == null) {
					logger.warn("Tracker[" + name + "] is a null reference.");
					continue;
				}
				try {
					tracker.registerContext(new TaskTrackerContextProxy(this, name));
					tracker.init();
					workingTrackers.put(name, tracker);
					trackersProperties.put(name, new TaskTrackerProperties(tracker.getCharacteristics()));
				} catch (Exception e) {
					logger.error("Initialization failed for Tracker[" + name + "].", e);
					workingTrackers.remove(name);
					trackersProperties.remove(name);
					continue;
				}
				logger.info("Tracker[" + name + "] initialized.");
			}
		}

		if (workingTrackers.isEmpty()) {
			logger.warn("TaskTrackerContext initialized, but no trackers are available.");
		} else {
			logger.info("TaskTrackerContext initialized successfully. " + workingTrackers.size()
					+ " tracker(s) available.");
		}
		dispatcherSemaphore = new Semaphore(1, false);
		waitingQueue = new LinkedBlockingQueue<FrameworkTrackedTask>(this.maxQueueSize);

		restoreRunningTasks();
		restoreWaitingQueue();

		Thread dispatcherThread = new Thread(new Runnable() {
			@Override
			public void run() {
				dispatcherWorker();
			}
		});
		dispatcherThread.start();

		shuttingdown = false;
		workingState = true;
	}

	/**
	 * 卸载
	 */
	public synchronized void destroy() {
		if (!workingState) {
			logger.warn("TaskTrackerContext has not been initialized. Calling destroy method has no effect.");
			return;
		}

		shuttingdown = true;
		notifyDispatcher();
		synchronized (shuttingdown) {
			try {
				shuttingdown.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		for (Map.Entry<String, ITaskTracker> trackerEntry : workingTrackers.entrySet()) {
			try {
				trackerEntry.getValue().destroy();
			} catch (Exception e) {
				logger.warn("Tracker[" + trackerEntry.getKey() + "] throws exception when destroying.", e);
			}
		}
		workingTrackers.clear();
		trackersProperties.clear();
		workingTrackers = null;
		trackersProperties = null;

		logger.info("TaskTrackerContext destroyed.");
		workingState = false;
	}

	private void restoreRunningTasks() {
		List<FrameworkTrackedTask> runningTasks = taskStoreProvider.getTasksByStatus(TaskStatus.RUNNING);
		for (FrameworkTrackedTask task : runningTasks) {
			ITaskTracker tracker = workingTrackers.get(task.getTrackerName());
			if (tracker == null) {
				logger.warn("Task[" + task.getGuid() + "] restore failed! Tracker[" + task.getTrackerName()
						+ "] not found.");
				taskStoreProvider.setResult(task.getGuid(), TaskStatus.LOST_TRACK, DateTimeHelper.getUtcDate(),
						TaskStatusMessage.TRACKER_NOT_FOUND,
						null);
				continue;
			}

			TaskTrackerProperties properties = trackersProperties.get(task.getTrackerName());
			if (properties.isTaskRestoreSupported()) {
				try {
					tracker.restoreTask(task.toTaskInfo());
				} catch (Exception e) {
					taskStoreProvider.setResult(task.getGuid(), TaskStatus.LOST_TRACK, DateTimeHelper.getUtcDate(),
							TaskStatusMessage.TRACKER_RESTORE_FAILED, null);
					continue;
				}
				properties.incrementConcurrent();
			} else {
				taskStoreProvider.setResult(task.getGuid(), TaskStatus.LOST_TRACK, DateTimeHelper.getUtcDate(),
						TaskStatusMessage.TRACKER_RESTORE_NOT_SUPPORTED, null);
				continue;
			}
		}
	}

	private void restoreWaitingQueue() {
		List<FrameworkTrackedTask> pendingTasks = taskStoreProvider.getTasksByStatus(TaskStatus.PENDING);
		if (autoRestoreWaitingQueue) {
			for (FrameworkTrackedTask task : pendingTasks) {
				if (!workingTrackers.containsKey(task.getTrackerName())) {
					taskStoreProvider.setResult(task.getGuid(), TaskStatus.ERROR, DateTimeHelper.getUtcDate(),
							TaskStatusMessage.TRACKER_NOT_FOUND, null);
				} else if (!waitingQueue.offer(task)) {
					taskStoreProvider.setResult(task.getGuid(), TaskStatus.ERROR, DateTimeHelper.getUtcDate(),
							TaskStatusMessage.WAITING_QUEUE_FULL, null);
				}
			}
		} else {
			List<String> guids = new ArrayList<String>(pendingTasks.size());
			for (FrameworkTrackedTask task : pendingTasks) {
				guids.add(task.getGuid());
			}
			taskStoreProvider.setResult(guids, TaskStatus.ERROR, DateTimeHelper.getUtcDate(),
					TaskStatusMessage.WAITING_QUEUE_FULL, null);
		}
	}

	public String newTask(String guid, String name, String trackerName, String owner, String inputPath,
			String outputPath, Map<String, String> properties) throws TrackerNotFoundException {
		if (!workingState)
			throw new RuntimeException("TaskTrackerContext has not been initialized.");

		ITaskTracker tracker = workingTrackers.get(trackerName);
		if (tracker == null)
			throw new TrackerNotFoundException("Tracker[" + trackerName + "] not found.");

		// 构造TaskInfo
		TaskInfo info = new TaskInfo(guid);
		info.setInputPath(inputPath);
		info.setOutputPath(outputPath);
		info.setStatus(TaskStatus.NONE);
		Map<String, String> target = info.getProperties();
		for (Map.Entry<String, String> prop : properties.entrySet()) {
			target.put(prop.getKey(), prop.getValue());
		}

		// 持久化保存
		FrameworkTrackedTaskInfo trackedTask = new FrameworkTrackedTaskInfo(info, name, trackerName, owner,
				DateTimeHelper.getUtcDate());
		taskStoreProvider.addTask(trackedTask);

		// 返回GUID
		return guid;
	}

	public void commitTask(String guid) throws WaitingQueueFullException {
		if (!workingState)
			throw new RuntimeException("TaskTrackerContext has not been initialized.");

		FrameworkTrackedTask trackedTask = taskStoreProvider.getTask(guid);
		if (trackedTask == null)
			throw new RuntimeException("guid not found!");

		commitTask(guid, trackedTask);
	}
	private void commitTask(String guid, FrameworkTrackedTask trackedTask) throws WaitingQueueFullException {
		TaskInfo info = trackedTask.toTaskInfo();

		// 加入队列
		synchronized (waitingQueue) {
			if (!waitingQueue.offer(trackedTask)) {
				throw new WaitingQueueFullException();
			}
			info.setStatus(TaskStatus.PENDING);
		}
		taskStoreProvider.checkAndUpdateStatus(guid, TaskStatus.PENDING, TaskStatus.NONE);

		// 通知任务启动
		notifyDispatcher();
	}

	private void notifyDispatcher() {
		dispatcherSemaphore.release();
	}

	private void dispatcherWorker() {
		List<FrameworkTrackedTask> list = new ArrayList<FrameworkTrackedTask>();
		FrameworkTrackedTask info;

		while (true) {
			try {
				dispatcherSemaphore.acquire();
			} catch (InterruptedException e) {
				logger.warn("waitingQueue.wait()", e);
				break;
			}

			if (shuttingdown)
				break;

			// 获得可用并发数
			int remainingConcurrentSize = this.getMaxConcurrentSize();
			if (remainingConcurrentSize > 0) {
				for (TaskTrackerProperties prop : trackersProperties.values()) {
					int rs = prop.getCurrentConcurrentSize();
					if (rs < 0) {
						remainingConcurrentSize = -1;
						break;
					} else {
						remainingConcurrentSize -= rs;
					}
				}
			}

			// 检查当前并发数
			int checkedCount = 0;
			synchronized (waitingQueue) {
				int queueSize = waitingQueue.size();
				while (checkedCount < queueSize && remainingConcurrentSize != 0
						&& (info = waitingQueue.poll()) != null) {
					TaskTrackerProperties properties = trackersProperties.get(info.getTrackerName());
					int eachRemainingSize = properties.getRemainingConcurrentSize();
					if (eachRemainingSize == 0) {
						waitingQueue.offer(info);
					} else {
						list.add(info);
						properties.incrementConcurrent();
						remainingConcurrentSize--;
					}
					checkedCount++;
				}
			}

			// 处理任务
			for (Iterator<FrameworkTrackedTask> iterator = list.iterator(); iterator.hasNext();) {
				info = iterator.next();

				ITaskTracker tracker = workingTrackers.get(info.getTrackerName());
				TaskInfo taskInfo = info.toTaskInfo();
				taskInfo.setProgress(new TaskProgressInfoProxy(this, info.getTrackerName(), info.getGuid()));

				logger.info("Scheduling Tracker[" + info.getTrackerName() + "]Task[" + info.getGuid() + "] to start.");
				taskStoreProvider.updateStartTime(info.getGuid(), DateTimeHelper.getUtcDate());
				try {
					tracker.startTask(taskInfo);
				} catch (Exception e) {
					logger.error("Task[" + info.getGuid() + "] failed to start.", e);
					trackersProperties.get(info.getTrackerName()).decrementConcurrent();
					continue;
				}
			}
			list.clear();
		}

		shuttingdown.notify();
	}

	public boolean cancelTask(String trackerName, String guid) throws Exception {
		ITaskTracker tracker = workingTrackers.get(trackerName);
		if (tracker == null)
			throw new RuntimeException("Tracker[" + trackerName + "] not found.");

		TaskTrackerProperties properties = trackersProperties.get(trackerName);
		if (!properties.isCancellationSupported())
			return false;

		try {
			tracker.cancelTask(guid);
		} catch (UnsupportedOperationException e) {
			return false;
		}

		return true;
	}

	/**
	 * 由TaskTracker调用，报告进度
	 * 
	 * @param tracker
	 * @param guid
	 * @param status
	 * @param info
	 */
	private void onStatusReport(String trackerName, String guid, int status, TaskResultInfo result) {
		logger.info("onStatusReport Tracker[" + trackerName + "]:Task[" + guid + "] status=" + status + " resultIsNull:"
				+ (result == null));

		ITaskTracker tracker = workingTrackers.get(trackerName);
		if (tracker == null)
			return;

		Date currentDateUtc = null;
		if (TaskStatus.isCompletedAnyway(status)) {
			currentDateUtc = DateTimeHelper.getUtcDate();
			TaskTrackerProperties properties = trackersProperties.get(trackerName);
			properties.decrementConcurrent();
		}

		if (result != null) {
			taskStoreProvider.setResult(guid, status, currentDateUtc, result.getPrimaryMessage(),
					result.getSecondaryMessage());
		} else {
			taskStoreProvider.setResult(guid, status, currentDateUtc, null, null);
		}
	}

	private void onProgressReport(String tracker, String guid, TaskProgressInfo progress) {
		taskStoreProvider.storeProgress(guid, progress);
	}

	public int getMaxQueueSize() {
		return maxQueueSize;
	}

	public void setMaxQueueSize(int maxQueueSize) {
		this.maxQueueSize = maxQueueSize;
	}

	public int getMaxConcurrentSize() {
		return maxConcurrentSize;
	}

	public void setMaxConcurrentSize(int maxConcurrentSize) {
		this.maxConcurrentSize = maxConcurrentSize;
	}

	public boolean isAutoRestoreWaitingQueue() {
		return autoRestoreWaitingQueue;
	}

	public void setAutoRestoreWaitingQueue(boolean autoRestoreWaitingQueue) {
		this.autoRestoreWaitingQueue = autoRestoreWaitingQueue;
	}

	public TaskDataStoreProvider getDataStoreProvider() {
		return dataStoreProvider;
	}

	public void setDataStoreProvider(TaskDataStoreProvider dataStoreProvider) {
		this.dataStoreProvider = dataStoreProvider;
	}

	public TaskStoreProvider getTaskStoreProvider() {
		return taskStoreProvider;
	}

	public void setTaskStoreProvider(TaskStoreProvider taskStoreProvider) {
		this.taskStoreProvider = taskStoreProvider;
	}

	public ITaskTrackerFactory[] getTrackers() {
		return this.trackers;
	}

	public void setTrackers(ITaskTrackerFactory... trackers) {
		this.trackers = trackers;
	}

	private static class TaskTrackerContextProxy implements ITaskTrackerContext {

		private TaskTrackerContext context;
		private String tracker;

		public TaskTrackerContextProxy(TaskTrackerContext context, String tracker) {
			this.context = context;
			this.tracker = tracker;
		}

		@Override
		public Logger getLogger() {
			return LoggerFactory.getLogger("TaskTracker[" + this.tracker + "]");
		}

		@Override
		public ITaskDataStore getTaskDataStore() {
			return new TaskDataStoreProxy(context.getDataStoreProvider(), this.tracker);
		}

		@Override
		public void reportStatus(String guid, int status, TaskResultInfo result) {
			context.onStatusReport(this.tracker, guid, status, result);
		}
	}

	private static class TaskProgressInfoProxy extends TaskProgressInfo {

		private TaskTrackerContext context;
		private String tracker;
		private String guid;

		public TaskProgressInfoProxy(TaskTrackerContext context, String tracker, String guid) {
			this.context = context;
			this.guid = guid;
		}

		@Override
		public void report() {
			context.onProgressReport(tracker, guid, this);
		}
	}
}
